package com.zh.amqp.rabbit.mqmsg;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import static com.zh.amqp.rabbit.mqmsg.MsgProducer.DEMO_QUEUE;

/**
 * @author ZH
 * @date 14:11 2020/8/19
 */
@Slf4j
@Component
public class MsgConsumer {

    /**
     * 监听MQ队列
     * ackMode：配置消费端手动签收消息
     */
    @RabbitListener(queues = {DEMO_QUEUE}, concurrency = "1")
    @RabbitHandler
    public void receiveMsg(@Payload MqMsg mqMsg, @Headers Map<String, Object> header, Channel channel) throws IOException{
        long deliveryTag = (long) header.get(AmqpHeaders.DELIVERY_TAG); //channel内消息的唯一标识

        log.info("【消费者 {} 】\t receive消息：uuid={}, deliveryTag={}, msg = {}",header.get("id"), header.get(AmqpHeaders.CONSUMER_TAG), deliveryTag, mqMsg);

        try {
            //处理消息 ...
            TimeUnit.SECONDS.sleep(2);
            char endChar = mqMsg.getContent().charAt(mqMsg.getContent().length()-1);
            if (endChar=='0'||endChar=='5'){
//                throw new RuntimeException("雨打五or零了 "+mqMsg.getContent());
            }

            //消息签收
            // 第一个参数是消息的唯一标识，
            // 第二个参数表示是否批量签收（即签收deliveryTag和before的msg）
            channel.basicAck(deliveryTag, true);
            log.info("【消费者 {} 】\t 已签收：deliveryTag={}",  header.get(AmqpHeaders.CONSUMER_TAG), deliveryTag);
        } catch (Exception ex) {
            //假设队列是头进尾出：
            // 第一个参数是消息的唯一标识，
            // 第二个参数表示是否批量签收（即签收deliveryTag和before的msg）
            // 第三个参数表示，消息是否重新进入队列（放入队尾），所以当只有一个consumer时，会死循环的。
            channel.basicNack(deliveryTag, false, false); //签收失败
//            channel.basicReject(deliveryTag, true); //拒绝签收,
//            channel.basicRecover(true);
            log.error("【消费者 {} 】\t 拒绝签收：deliveryTag={}，操作失败：cause = {}", header.get(AmqpHeaders.CONSUMER_TAG), deliveryTag, ex.getMessage());

            //区别：
            //channel.basicNack(); 可以一次拒绝多条消息
            //channel.basicReject(); 一次只能拒绝一条消息

            //一般的业务处理：消息设为不重回队列尾部，而是记录日志、通知告警、记录错误表手动补偿
            channel.basicNack(deliveryTag, false, false);
            
        }
    }

}

